package com.kamistoat.rocketmqlearning.msgProducer;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * 官方建议复用Producer，而不是频繁创建与销毁。因此使用Bean来管理.
 * 但务必注意，当Bean的模式选择为单例模式时，Producer的start()和shutdown()方法此时需要由Bean的生命周期来管理，而不能手动执行。
 */
@Configuration
public class MsgProducerConfig {
    /**
     * 通用的生产者Bean。如果存在其他业务方的Producer，继续写Bean即可，通过Bean的name来进行区分
     *
     * @return DefaultMQProducer
     */
    @Bean(name = "msgProducer", initMethod = "start", destroyMethod = "shutdown")
    public DefaultMQProducer msgProducer() {
        /**
         * 直接使用RocketMQ的库方法创建默认生产者。
         */
        DefaultMQProducer producer = new DefaultMQProducer();

        /**
         * 设置生产者组的名称。
         * 默认值为 DEFAULT_PRODUCER。
         * 同一个
         */
        producer.setProducerGroup("producerGroup");

        /**
         * 当发送的消息目标Topic不存在时，Broker会自动创建。设置创建的Topic的初始队列数
         * 默认值为4
         */
        producer.setDefaultTopicQueueNums(4);

        /**
         * 设置发送消息时的超时时间。单位毫秒
         * 默认值为10000
         */
        producer.setSendMsgTimeout(10000);

        /**
         * 设置消息Body超过多大开始压缩(Consumer收到消息会自动解压缩)
         * 默认4096
         */
        producer.setCompressMsgBodyOverHowmuch(4096);

        /**
         * 如果发送消息返回sendResult，但是sendStatus!=SEND_OK，是否重试向其他Broker重复发送
         * 默认false
         */
        producer.setRetryAnotherBrokerWhenNotStoreOK(false);

        /**
         * 如果消息发送失败，最大重试次数。只对同步发送生效
         * 默认为2
         */
        producer.setRetryTimesWhenSendFailed(2);

        /**
         * 如果消息发送失败，最大重试次数。只对异步发送生效
         * 默认为2
         */
        producer.setRetryTimesWhenSendAsyncFailed(2);

        /**
         * 指定NameServer地址。如果NameServer是集群则需使用;指定链接所有的NameServer
         * 生产者会随机链接一个NameServer进行后序流程
         */
        producer.setNamesrvAddr("172.19.0.20:9876;172.19.0.21:9877");

        return producer;
    }

    /**
     * 发送事务消息的生产者Bean
     *
     * @return TransactionMQProducer
     */
    @Bean(name = "transactionMsgProducer", initMethod = "start", destroyMethod = "shutdown")
    public TransactionMQProducer transactionMsgProducer() {
        TransactionMQProducer producer = new TransactionMQProducer();
        producer.setProducerGroup("transProducerGroup");
        producer.setDefaultTopicQueueNums(4);
        producer.setSendMsgTimeout(10000);
        producer.setCompressMsgBodyOverHowmuch(4096);
        producer.setRetryAnotherBrokerWhenNotStoreOK(false);
        producer.setRetryTimesWhenSendFailed(2);
        producer.setRetryTimesWhenSendAsyncFailed(2);
        producer.setNamesrvAddr("172.19.0.20:9876;172.19.0.21:9877");
        /**
         * 创建线程池用于事务状态回查。TransactionMQProducer必需一个线程池，如果不手动创建也会默认生成一个
         */
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("client-transaction-msg-check-thread");
                        return thread;
                    }
                }
        );
        /**
         * 绑定用于事务状态回查的线程池。如不手动指定也会默认生成一个
         */
        producer.setExecutorService(threadPoolExecutor);

        /**
         * 绑定事务监听器。需要是继承TransactionListener接口的类。
         */
        producer.setTransactionListener(new TransactionListener() {
            /**
             * 预提交成功回调接口。生产者向MQ预提交半事务消息成功后，调用该方法执行内部业务。
             * 通常可以将内部业务设置为 本地事务/分布式事务
             * @param msg 投递的消息
             * @param arg 投递消息时传递的参数
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                /**
                 * 可以调用其他服务层代码完成事务操作。
                 */
                System.out.println("Broker消息预提交成功");
                String tags = msg.getTags();
                /**
                 * 模拟事务执行失败或者由于宕机造成的UNKNOW
                 */
                if (StringUtils.equals(tags, "TagD")) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if (StringUtils.equals(tags, "TagE")) {
                    return LocalTransactionState.UNKNOW;
                } else {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            }

            /**
             * 事务状态回查接口
             * @param msg 投递的消息
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("执行消息回查" + msg.getTags());
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        return producer;
    }
}
